package com.bw.gmall.realtime.Day0914;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TableProcess;
import com.bw.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

public class TableProcessFunction2 extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private Connection connection;

    private  MapStateDescriptor<String, TableProcess> mapstate;

    public TableProcessFunction2(MapStateDescriptor<String, TableProcess> mapstate) {
        this.mapstate = mapstate;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
      //phoenix不需要用户密码
         connection= DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    /*
    value:
    {
    "database":"gmall",
    "xid":18944,
    "data":{"login_name":"1","nick_name":"zs","id":1602},
    "commit":true,"type":"insert",
    "table":"user_info",
    "ts":1592813677
     "sinkTable":dim_user_info}


     insert    dim_user_info(login_name,nick_name,id)  values("1","zs","1602)


    * */
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> map = ctx.getBroadcastState(mapstate);
//        System.out.println(map.get("activity_info"));
//        System.out.println(map.get("user_info"));
//        System.out.println(map.get("sku_info"));
//        System.out.println(value);
//        out.collect(value);
        String table = value.getString("table");
        TableProcess t = map.get(table);
        if(t!=null){
            System.out.println("处理之前数据:"+value);
            filterCoulums(value.getJSONObject("data"),t.getSink_columns());
            System.out.println("处理之前数据:"+value);//{id,name,sinkTable}
            value.put("sinkTable",t.getSink_table());
            out.collect(value);//输出   变名 dim_user_info    name,id

        }else {
            System.out.println("此数据不是维度数据表名为:"+table);
        }
    }

    /*
    CDC主流:>
     {"before":null
     "after":{"source_table":"base_trademark","sink_table":"dim_base_trademark","sink_columns":"id,tm_name",
     "sink_pk":"id","sink_extend":null},
     "source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source",
     "ts_ms":1721274453437,"snapshot":"false",
     "db":"gmall_config","sequence":null,
     "table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},
     "op":"r"  "op":"c",
     "ts_ms":1721274453437,"transaction":null}

     */
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {

        //1.获取数据  创建表
        JSONObject jsonObject = JSON.parseObject(value);
        //2.
        TableProcess tp = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);
        checkTable(tp.getSink_table(),tp.getSink_columns(),tp.getSink_pk(),tp.getSink_extend());
        //2.创建完表之后要不要把 把 维度表存入状态中给主流
        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(mapstate);
        state.put(tp.getSource_table(),tp);
    }
    //sink_table  表名     sink_columns 字段   主键rowkey    约束sink_extend
    private void checkTable(String sink_table, String sink_columns, String sink_pk, String sink_extend)  {


        /*
         * 1.获取连接（连接池）
         * 2.获取ps
         * 3.发送sql
         * 4.执行sql
         *
         * */


        PreparedStatement ps=null;



        try{
            //处理字段
            if(sink_pk==null||"".equals(sink_pk)){
                sink_pk="id";
            }
            if (sink_extend==null){
                sink_extend="";
            }
            //拼接sql
            StringBuilder createTable = new StringBuilder("CREATE  TABLE  IF  NOT EXISTS    ")
                    .append(GmallConfig.HBASE_SCHEMA)   //not  exists    schema "GMALL_REALTIME"
                    .append(".")
                    .append(sink_table)
                    .append("(");

            String[] split = sink_columns.split(",");

            for (int i = 0; i < split.length; i++) {
                //取出字段

                //是否为主键
                if(sink_pk.equals(split[i])){
                    createTable.append(split[i]).append("  varchar PRIMARY KEY");
                }else{
                    createTable.append(split[i]).append("  varchar");
                }
                //判断是否是最后一个字段
                if(i<split.length-1){
                    createTable.append(",");
                }

            }
            createTable.append(")").append(sink_extend);

            System.out.println("创建表语句为:"+createTable);

            String sql = createTable.toString();




            ps = connection.prepareStatement(sql);
            ps.execute();


            //执行sql //正常数据
        }catch (Exception e){
            //把编译时异常  变成运行时异常（程序不会中断）
            new RuntimeException("建表失败:"+sink_table);
        }finally {
            if(ps!=null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

        }



    }
    private void filterCoulums(JSONObject data, String sink_columns) {
        String[] split = sink_columns.split(",");
        //把数组转换成列表
        List<String> list = Arrays.asList(split);
//        "data":{"id":12,"tm_name":"bw","logo_url":"/static/bw.jpg"}
//         id,tm_name
        //对数据进行遍历  把json当中map
        Set<Map.Entry<String, Object>> entries = data.entrySet();



        //迭代器
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, Object> next = iterator.next();
            //这个字符就是包含的意思
            //如果把字符串转换成list    那么就判断是否包含元素
            if(!list.contains(next.getKey())){
                //不包含的话把key-value删除掉
                iterator.remove();
            }
        }

    }

}

